package com.atguigu.userprofile.app

import java.util.Properties

import com.atguigu.userprofile.common.bean.{TagInfo, TaskInfo, TaskTagRule}
import com.atguigu.userprofile.common.constants.ConstCode
import com.atguigu.userprofile.common.dao
import com.atguigu.userprofile.common.dao.{TagInfoDAO, TaskInfoDAO, TaskTagRuleDAO}
import com.atguigu.userprofile.common.util.{MyPropertiesUtil, MySqlUtil}
import org.apache.spark.SparkConf
import org.apache.spark.sql.SparkSession

object TaskSqlApp {


//  1  读取标签的信息
//    从mysql 中读取 tag_info   task_info   task_tag_rule
//
//  2  根据标签的定义、规则 读取数仓的表
  //  查询的目标
//    select    根据之前的定义，拼接组合成一条select 语句
//
//  3  把数据写入到指定的画像库中的表
//     1 要定义 标签表的结构
//     2 用insert 把数据写入到 标签表中

  def main(args: Array[String]): Unit = {

    //注意 要部署到yarn环境时 要注掉.setMaster
    val sparkConf: SparkConf = new SparkConf().setAppName("task_sql_app")//.setMaster("local[*]")
    val sparkSession: SparkSession = SparkSession.builder().config(sparkConf).enableHiveSupport().getOrCreate()

    //  1  读取标签的信息
    //    从mysql 中读取 tag_info   task_info   task_tag_rule
    //    // 读1条tag_info 还是多个？
    //主函数执行一次 是处理一个标签 还是处理多个标签 ？
    // 执行一次 只处理一个标签
    //  但是所有sql类的标签任务 都使用该类进行处理

     val taskId: String = args(0)
     val taskDate: String = args(1)

     val tagInfo: TagInfo = TagInfoDAO.getTagInfoByTask(taskId)
     val taskInfo: TaskInfo = TaskInfoDAO.getTaskInfo(taskId)
    val taskTagRuleList: List[TaskTagRule] = TaskTagRuleDAO.getTaskTagRuleList(taskId)
    println(tagInfo)
    println(taskInfo)
    println(taskTagRuleList)

    //  2  根据标签的定义、规则 读取数仓的表
    //  查询的目标    标签表（一个标签一个表） ：  uid  ， tag_value
    //    select    根据之前的定义，拼接组合成一条select 语句
    //
    // 因为 F M U 不是最终想要写入标签的数据 ，以task_tag_rule映射的 四级标签值为准
    //  要以task_tag_rule映射 生成一个case when子句
    //  case 字段  when  值1  then 值2 end
    var tagvalueSQL:String=null
    if(taskTagRuleList.size>0){  //针对文本型 值  有四级标签
      val whenThenList: List[String] = taskTagRuleList.map { taskTagRule =>
        s" when  '${taskTagRule.queryValue}'  then  '${taskTagRule.subTagValue}' "
      }
      val whenThenSQL: String = whenThenList.mkString(" ")
       tagvalueSQL =s"case query_value  $whenThenSQL end "
    }else{ //针对数字值 日期  无四级标签
       tagvalueSQL =s"  query_value   "
    }


   //把sql中的日期变量转为任务日期
   val taskSql: String = taskInfo.taskSql.replace("$dt",taskDate)

    val selectSQL=s" select uid , $tagvalueSQL as tag_value from (${taskSql}) tsql "
    println(selectSQL)

    //  3  把数据写入到指定的画像库中的表
    //     1 要定义 标签表的结构
    val tableName=tagInfo.tagCode.toLowerCase

    val tagValueType= tagInfo.tagValueType match {
      case ConstCode.TAG_VALUE_TYPE_STRING => "STRING"
      case ConstCode.TAG_VALUE_TYPE_LONG => "BIGINT"
      case ConstCode.TAG_VALUE_TYPE_DECIMAL => "DECIMAL(16,2)"
      case ConstCode.TAG_VALUE_TYPE_DATE => "STRING"
    }

     val properties: Properties = MyPropertiesUtil.load("config.properties")
    val hdfsPath: String = properties.getProperty("hdfs-store.path")
    val dwName: String = properties.getProperty("data-warehouse.dbname")
    val upName: String = properties.getProperty("user-profile.dbname")

    val  createTableSQL=
      s"""
         | create table if not exists $upName.$tableName
         |( uid string ,tag_value $tagValueType )
         |comment '${tagInfo.tagName}'
         |partitioned by ( dt string)
         |ROW FORMAT DELIMITED FIELDS TERMINATED BY '\\t'
         | location '$hdfsPath/$upName/$tableName'
       """.stripMargin

    println(createTableSQL)
    sparkSession.sql(createTableSQL)

    //     2 用insert 把数据写入到 标签表中
    sparkSession.sql(s"use $dwName")
   val insertSQL=
     s"""
        | insert overwrite table $upName.$tableName partition (dt= '$taskDate')
        | $selectSQL
        |
      """.stripMargin

    println(insertSQL)
    sparkSession.sql(insertSQL)
  }

}
